package com.doitedu.core

import com.doitedu.utils.SparkUtil
import org.apache.spark.sql.execution.SparkPlan

import java.sql.{Connection, DriverManager, PreparedStatement}

/**
 * @Date: 22.6.29 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 */
object C13_Data2Mysql {
  def main(args: Array[String]): Unit = {
    val sc = SparkUtil.getSparkContext("data  2  mysql")
    val rdd1 = sc.textFile("data/stu")
    val rdd2 = rdd1.map(line => {
      val arr = line.split(",")
      (arr(0).toInt, arr(1), arr(2), arr(3).toDouble)
    })
    // 成绩大于60  符合条件的数据会被留下
    //    val rdd3 = rdd2.filter()
    // 过滤出成绩大于60  有名字的数据
    val rdd4 = rdd2.filter(stu=>stu._2.trim != "")

     // 将数据存储到mysql
/*    rdd4.foreach(stu=>{  // 没有执行
      // 在每个分区内部 , 每条数据执行一次函数
      // 每条数据获取一次连接
      val conn: Connection =  DriverManager.getConnection("jdbc:mysql://localhost:3306/cool?characterEncoding=UTF8" ,"root","root")
      val ps: PreparedStatement = conn.prepareStatement("insert  into  tb_score values(?,?,?,?)")
      ps.setInt(1,stu._1)
      ps.setString(2,stu._2)
      ps.setString(3,stu._3)
      ps.setDouble(4 , stu._4)
      ps.executeUpdate()
    })*/

    // 每个分区执行一次的方法   mapPartitions

    rdd4.foreachPartition(iter=>{
      // iter 一个分区的数据
      // 每个分区执行一次 获取一次连接
      val conn: Connection =  DriverManager.getConnection("jdbc:mysql://localhost:3306/cool?characterEncoding=UTF8" ,"root","root")
      val ps: PreparedStatement = conn.prepareStatement("insert  into  tb_score values(?,?,?,?)")
      //处理分区内的每条数据
      iter.foreach(stu=>{
        ps.setInt(1,stu._1)
        ps.setString(2,stu._2)
        ps.setString(3,stu._3)
        ps.setDouble(4 , stu._4)
        ps.executeUpdate()
      })
    })


    sc.stop()
  }
}
